构建基于Serverless架构的弹性高可用音视频处理系统

简单视频处理系统

前提条件

• 函数计算：开通函数计算服务。

• 对象存储OSS：控制台创建存储空间。

操作步骤

下列步骤演示如何搭建一个Serverless视频转换系统。当视频上传到OSS Bucket的source目录时会触发函数。函数将MP4格式的视频文件转为FLV格式，并将FLV格式视频文件保存在OSS Bucket下的dest目录。

1. 创建函数。

   在创建函数页面，单击事件函数，配置以下项，然后单击创建。

   • 基本设置：配置函数的基本信息，包括函数名称和地域。您可以选择距离自己较近的区域，如华东1（杭州）。

   • 函数代码：函数的运行环境选择Python 3.10，代码上传方式选择使用示例代码。

   • 高级配置：由于处理视频文件比较耗时，并且要考虑视频文件的大小，所以本文选择vCPU规格为4核，内存规格为8 GB，临时硬盘大小为10 GB，执行超时时间为7200秒。函数角色请选择AliyunFcDefaultRole。

   其他配置项使用默认值，关于创建函数的详细配置信息，请参见创建函数。

2. 创建OSS触发器。

   1. 在函数详情页面，选择配置页签，在左侧导航栏，单击触发器，然后单击创建触发器。

   2. 在创建触发器面板，触发器类型选择双向集成触发器类型中的对象存储OSS，设置其余配置项，然后单击确定。

      配置项	操作	本文示例
      名称	填写自定义的触发器名称。	oss-trigger
      版本或别名	默认值为LATEST，如果您需要创建其他版本或别名的触发器，需先在函数详情页的版本或别名下拉列表选择该版本。关于版本和别名的简介，请参见版本管理和别名管理。	LATEST
      Bucket 名称	选择已创建的OSS Bucket。	testbucket
      文件前缀	输入要匹配的文件名称的前缀。建议您配置文件前缀和后缀，避免触发事件嵌套循环触发引起额外费用。此外，一个Bucket的不同触发器如果指定了相同的事件类型，则前缀和后缀不能重复。详细信息，请参见OSS触发器触发规则。 重要 文件前缀不能以/开头，否则会导致OSS触发器无法被触发。	source
      文件后缀	输入要匹配的文件名称的后缀。强烈建议您配置前缀和后缀，避免触发事件嵌套循环触发引起额外费用。另外，一个Bucket的不同触发器如果指定了相同的事件类型，则前缀和后缀不能重复。详细信息，请参见OSS触发器触发规则。	mp4
      触发事件	选择一个或多个触发事件。关于对象存储OSS的事件类型，请参见OSS事件定义。 本示例选择oss:ObjectCreated:PutObject。	oss:ObjectCreated:PutObject，oss:ObjectCreated:PostObject，oss:ObjectCreated:CompleteMultipartUpload
      角色名称	选择AliyunOSSEventNotificationRole。 说明 如果您第一次创建该类型的触发器，则需要在单击确定后，在弹出的对话框中选择立即授权。	AliyunOSSEventNotificationRole

      重要

      请注意一个 Bucket 的一个前缀下只能支持配置一个OSS触发器，如果配置多个会出现Cannot specify overlapping prefix and suffix with same event type 的报错。可以选择以下任一方法解决该问题。

      • 将对象存储存储桶名设置为另一个Bucket名称。

      • 更改前缀名称。

      • 删除冲突的OSS触发器。

3. 编写函数代码。

   1. 在函数详情页面，单击函数代码页签，在代码编辑器中编写代码。

      本文以Python语言为例，示例代码如下。

      # -*- coding: utf-8 -*-
      import logging
      import oss2
      import os
      import json
      import subprocess
      import shutil
      
      logging.getLogger("oss2.api").setLevel(logging.ERROR)
      logging.getLogger("oss2.auth").setLevel(logging.ERROR)
      LOGGER = logging.getLogger()
      
      
      def get_fileNameExt(filename):
          (_, tempfilename) = os.path.split(filename)
          (shortname, extension) = os.path.splitext(tempfilename)
          return shortname, extension
      
      
      def handler(event, context):
          LOGGER.info(event)
          evt = json.loads(event)
          evt = evt["events"]
          oss_bucket_name = evt[0]["oss"]["bucket"]["name"]
          object_key = evt[0]["oss"]["object"]["key"]
          
          output_dir = "dest"
          dst_format = "flv"
          shortname, _ = get_fileNameExt(object_key)
          creds = context.credentials
          auth = oss2.StsAuth(creds.accessKeyId,
                              creds.accessKeySecret, creds.securityToken)
          oss_client = oss2.Bucket(auth, 'oss-%s-internal.aliyuncs.com' %
                                   context.region, oss_bucket_name)
      
          exist = oss_client.object_exists(object_key)
          if not exist:
              raise Exception("object {} is not exist".format(object_key))
      
          input_path = oss_client.sign_url('GET', object_key, 6 * 3600)
          # m3u8 特殊处理
          rid = context.request_id
          if dst_format == "m3u8":
              return handle_m3u8(rid, oss_client, input_path, shortname, output_dir)
          else:
              return handle_common(rid, oss_client, input_path, shortname, output_dir, dst_format)
      
      
      def handle_m3u8(request_id, oss_client, input_path, shortname, output_dir):
          ts_dir = '/tmp/ts'
          if os.path.exists(ts_dir):
              shutil.rmtree(ts_dir)
          os.mkdir(ts_dir)
          transcoded_filepath = os.path.join('/tmp', shortname + '.ts')
          split_transcoded_filepath = os.path.join(
              ts_dir, shortname + '_%03d.ts')
          cmd1 = ['ffmpeg', '-y', '-i', input_path, '-c:v',
                  'libx264', transcoded_filepath]
      
          cmd2 = ['ffmpeg', '-y', '-i', transcoded_filepath, '-c', 'copy', '-map',  '0',  '-f', 'segment',
                  '-segment_list', os.path.join(ts_dir, 'playlist.m3u8'), '-segment_time', '10', split_transcoded_filepath]
      
          try:
              subprocess.run(
                  cmd1, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
      
              subprocess.run(
                  cmd2, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
      
              for filename in os.listdir(ts_dir):
                  filepath = os.path.join(ts_dir, filename)
                  filekey = os.path.join(output_dir, shortname, filename)
                  oss_client.put_object_from_file(filekey, filepath)
                  os.remove(filepath)
                  print("Uploaded {} to {}".format(filepath, filekey))
      
          except subprocess.CalledProcessError as exc:
              # if transcode fail，trigger invoke dest-fail function
              raise Exception(request_id + " transcode failure, detail: " + str(exc))
      
          finally:
              if os.path.exists(ts_dir):
                  shutil.rmtree(ts_dir)
      
              # remove ts 文件
              if os.path.exists(transcoded_filepath):
                  os.remove(transcoded_filepath)
      
          return {}
      
      
      def handle_common(request_id, oss_client, input_path, shortname, output_dir, dst_format):
          transcoded_filepath = os.path.join('/tmp', shortname + '.' + dst_format)
          if os.path.exists(transcoded_filepath):
              os.remove(transcoded_filepath)
          cmd = ["ffmpeg", "-y", "-i", input_path, transcoded_filepath]
          try:
              subprocess.run(
                  cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
      
              oss_client.put_object_from_file(
                  os.path.join(output_dir, shortname + '.' + dst_format), transcoded_filepath)
          except subprocess.CalledProcessError as exc:
              # if transcode fail，trigger invoke dest-fail function
              raise Exception(request_id + " transcode failure, detail: " + str(exc))
          finally:
              if os.path.exists(transcoded_filepath):
                  os.remove(transcoded_filepath)
          return {}
      

   2. 单击部署代码。

4. 测试函数代码。

   您可以通过配置函数入口参数模拟OSS事件参数进行测试，验证代码的正确性。在实际操作过程中，当发生OSS事件时，函数会自动被触发执行。

   1. 在函数详情页面，单击函数代码页签，然后单击测试函数右侧图标，从下拉列表中，选择配置测试参数。

   2. 在配置测试参数面板，选择创建新测试事件或编辑已有测试事件页签，填写事件名称和事件内容。然后单击确定。

      本文配置的事件模板如下。请将Bucket名称、ETag、视频文件名替换成实际的值。

      {
          "events": [
              {
                  "eventName": "oss:ObjectCreated:CompleteMultipartUpload",
                  "eventSource": "acs:oss",
                  "eventTime": "2022-08-13T06:45:43.000Z",
                  "eventVersion": "1.0",
                  "oss": {
                      "bucket": {
                          "name": "testbucket"
                      },
                      "object": {
                          "eTag": "688A7BF4F233DC9C88A80BF985AB****",
                          "key": "source/a.mp4"
                      }
                  },
                  "region": "cn-hangzhou"
              }
          ]
      }

      说明

      ETag (entity tag) 在每个Object生成的时候被创建，用于标识一个Object的内容。ETag可以在OSS的文件列表中，点击目标对象的详情获取。

      

      关于event参数具体信息，请参见配置函数入口参数。

   3. 单击测试函数。函数执行成功后，可以在目标Bucket的dest/路径下找到转换后的视频文件。

      

视频处理工作流系统

前提条件

• 开通服务并创建Bucket

  • 函数计算：开通函数计算服务。

  • 对象存储OSS：控制台创建存储空间。

  • 云工作流：开通云工作流。

  • 文件存储 NAS：开通文件存储服务NAS。

  • 专有网络 VPC：开通专有网络VPC。

• 配置函数角色

  • AliyunFcDefaultRole：创建函数时需要配置该角色，同时该角色需要配置权限AliyunFCInvocationAccess、AliyunFnFFullAccess、AliyunOSSFullAccess，用于调用函数、管理工作流和管理对象存储OSS。

  • AliyunOSSEventNotificationRole：OSS默认使用此角色发送事件通知。

  • fnf-execution-default-role：创建及管理工作流所需要的角色，该角色需要配置权限AliyunFCInvocationAccess和AliyunFnFFullAccess。

    请在该角色的信任策略中为工作流服务授予AssumeRole权限，如下所示。

    {
      "Statement": [
        {
          "Action": "sts:AssumeRole",
          "Effect": "Allow",
          "Principal": {
            "Service": [
              "fnf.aliyuncs.com"
            ]
          }
        }
      ],
      "Version": "1"
    }

• Serverless Devs工具安装及配置

  • 安装Serverless Devs工具及依赖

  • 配置Serverless Devs

操作步骤